package com.atguigu.flink.java.chapter_7;

import com.atguigu.flink.java.chapter_5.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/12 20:29
 */
public class Flink13_Chapter07_SideOutput_Split {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        System.out.println(env.getConfig());

        SingleOutputStreamOperator<WaterSensor> result =
          env
            .socketTextStream("hadoop102", 9999)  // 在socket终端只输入毫秒级别的时间戳
            .map(new MapFunction<String, WaterSensor>() {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] datas = value.split(",");
                    return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));

                }
            })
            .keyBy(ws -> ws.getTs())
            .process(new KeyedProcessFunction<Long, WaterSensor, WaterSensor>() {
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                    out.collect(value);
                    if (value.getVc() > 5) { //水位大于5的写入到侧输出流
                        ctx.output(new OutputTag<WaterSensor>("警告") {}, value);
                    }
                }
            });

        result.print("主流");
        result.getSideOutput(new OutputTag<WaterSensor>("警告"){}).print("警告");

        env.execute();
    }
}
